package com.ssm.common.util.thread;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.concurrent.*;

public class KafkaThreadUtil {
    private static final ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("kafka-Thread-%d").setDaemon(true).build();
    private static final ExecutorService executorService = new ThreadPoolExecutor(5, 10,0L
            , TimeUnit.MILLISECONDS,new LinkedBlockingDeque<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());

    private static ThreadFactory namedScheduleThreadFactory = new ThreadFactoryBuilder().setNameFormat("kafka-Schedule-%d").setDaemon(true).build();

    private static ScheduledExecutorService scheduledExecutor = new ScheduledThreadPoolExecutor(1, namedScheduleThreadFactory);

    /**
     * 提交任务
     */
    public static <T> Future<T> submit(Callable<T> callable) {
        return executorService.submit(callable);
    }

    /**
     * 提交任务
     */
    public static void submit(Runnable runnable) {
        executorService.submit(runnable);
    }

    /**
     * 提交周期任务 不能抛异常，要catch {@link Throwable}
     */
    public static void schedule(Runnable callable, long delay, long period, TimeUnit unit) {
        scheduledExecutor.scheduleAtFixedRate(callable, delay, period, unit);
    }

    public static void shutdown() {
        executorService.shutdown();
        scheduledExecutor.shutdown();
    }

    public static void shutdownNow() {
        executorService.shutdownNow();
        scheduledExecutor.shutdownNow();
    }
}
